package com.alibaba.otter.node.etl.load.loader.mq.rabbit.provider;

import com.alibaba.otter.node.etl.load.loader.mq.rabbit.consume.Handler;
import com.google.common.cache.CacheBuilder;
import com.google.common.cache.CacheLoader;
import com.google.common.cache.LoadingCache;
import com.rabbitmq.client.AMQP;
import lombok.NoArgsConstructor;
import lombok.extern.slf4j.Slf4j;

import java.lang.reflect.ParameterizedType;
import java.lang.reflect.Type;
import java.util.concurrent.ExecutionException;

@Slf4j
@NoArgsConstructor
public class DefaultTargetClassExplorer implements TargetClassExplorer {

    public DefaultTargetClassExplorer(Handler<?> handler) {
        this.handler = handler;
    }

    private Handler<?> handler;

    private static final Class<String> defaultClz = String.class;

    private static LoadingCache<String, Class> clzCache = CacheBuilder.newBuilder()
        .build(new CacheLoader<String, Class>() {
            @Override
            public Class load(String key) throws Exception {
                try {
                    return Class.forName(key);
                } catch (ClassNotFoundException e) {
                    log.warn("无此消息类型: " + key + "，将反序列化为String类型。", e);
                    return defaultClz;
                }
            }
        });

    @Override
    public Class explore(AMQP.BasicProperties properties) {
        Class result = findClassFromGenericType(handler);
        result = result == null ? findClassFromAMQPProperties(properties) : result;
        result = result == null ? defaultClz : result;
        return result;
    }

    private Class findClassFromGenericType(Handler handler) {
        Class result;
        Type[] genericInterfaces = handler.getClass().getGenericInterfaces();
        for (Type genericInterface : genericInterfaces) {
            if (!(genericInterface instanceof ParameterizedType)) {
                continue;
            }
            ParameterizedType type = ((ParameterizedType) genericInterface);
            Class<?> interfaceClz = forClz(type.getRawType().getTypeName());
            if (interfaceClz == null || !interfaceClz.isAssignableFrom(Handler.class)) {
                continue;
            }
            Type[] genericTypes = type.getActualTypeArguments();
            for (Type genericType : genericTypes) {
                result = forClz(genericType.getTypeName());
                if (result != null) {
                    return result;
                }
            }
        }
        return null;
    }

    private Class<?> findClassFromAMQPProperties(AMQP.BasicProperties properties) {
        if (properties.getType() != null) {
            try {
                return clzCache.get(properties.getType());
            } catch (ExecutionException e) {
                log.error("failed to detected deserialize type.", e);
            }
        }
        return null;
    }

    private Class<?> forClz(String clzName) {
        try {
            return Class.forName(clzName);
        } catch (ClassNotFoundException e) {
            log.warn(e.getMessage(), e);
            return null;
        }
    }
}
